
const { resolve, reject, some } = require('bluebird');
const kafka = require('kafka-node')
const Client = kafka.KafkaClient;
const Offset = kafka.Offset;
const Consumer = kafka.Consumer;
var utils = require('../../utils');
const deviceUtils = require('../device-utils');
const LocalStorage = require('node-localstorage').LocalStorage;
const localStorage = new LocalStorage('./localStorage');
function KafkaClient(_data, _logger, _events) {
    var data = _data;                   // Current webapi data
    var logger = _logger;               // Logger var working = false;                // Working flag to manage overloading polling and connection
    var working = false;                // Working flag to manage overloading polling and connection
    var connected = false;              // Connected flag
    var events = _events;               // Events to commit change to runtime
    var lastStatus = '';                // Last webapi status
    var varsValue = [];                 // Signale to send to frontend { id, type, value }
    var requestItemsMap = {};           // Map of request (JSON, CSV, XML, ...) {key: item path, value: tag}
    var overloading = 0;                // Overloading counter to mange the break connection
    var lastTimestampRequest;           // Last Timestamp request
    var lastTimestampValue;             // Last Timestamp of asked values
    var newItemsCount;
    var client;
    var consumer;
    var host;
    var allData;
    var apiProperty = { getTags: null, postTags: null, format: 'JSON', ownFlag: true };
    this.connect = function () {
        return new Promise((resolve, reject) => {
            if (_checkConnection()) {
                try {
                    if (_checkWorking(true)) {
                        const property = getConnectionOptions(data.property)
                        host = !!property.url ? property.url : "10.39.64.34:9092"
                        client = new Client({ kafkaHost: host });
                        client.on('ready', function () {
                            logger.info('kafka connect-ok')
                            _emitStatus('connect-ok');
                            _checkWorking(false);
                            resolve();
                        })

                        client.on('error', function () {
                            _emitStatus('connect-ok');
                        })
                    }
                } catch (error) {
                    logger.error(`'${data.name}' try to connect error! ${err}`);
                    _checkWorking(false);
                    connected = false;
                    _emitStatus('connect-error');
                    _clearVarsValue();
                    reject();
                }
            }


        })
    }

    this.getDeviceKafkaResult = function () {
        return new Promise(function (resolve, reject) {
            const options = {
                //  自动提交配置   (false 不会提交偏移量，每次都从头开始读取)
                autoCommit: true,
                autoCommitIntervalMs: 5000,
                //  如果设置为true，则consumer将从有效负载中的给定偏移量中获取消息
                fromOffset: false,
                timeout: 5000,
                interval: 10000
            };

            const topics = [
                {
                    topic: "data_iot_device_measure",
                    partition: 0,
                    offset: 0,
                },
            ];
            const property = getConnectionOptions(data.property)
            host = !!property.url ? property.url : "10.39.64.34:9092"
            client = new Client({ kafkaHost: host });

            const consumer = new Consumer(client, topics, options);

            consumer.on('message', (data) => {
                data.value = JSON.parse(data.value)
                const sendVal = _convertData(data.value)
                localStorage.setItem('consumerMsg', JSON.stringify(sendVal))
                resolve(sendVal)
            })
            consumer.on('error', function (error) {
                reject(error)
                logger.error('Error connecting to Kafka:', error);
            });
        })
    }

    const _convertData = function(data){
        const obj = {}
        const joinKey = data.devId
        for(let key in data){
            
            if(Array.isArray(data[key])){
                data[key].forEach(item => {
                    obj['t_' + joinKey + '_' + item.metric] = item.value
                })
            }else{
                obj[key] = data[key]
            }
        }
        return obj
    }


    this.polling = function () {
        let obj = []
        if (localStorage.getItem('consumerMsg')) {
            obj = JSON.parse(localStorage.getItem('consumerMsg'));
        }
        if (_checkWorking(true)) {
            if (client) {
                try {
                    var varsValueChanged = _updateVarsValue(obj);
                    lastTimestampValue = new Date().getTime();
                    _emitValues(varsValue);

                    if (this.addDaq) {
                        this.addDaq(varsValueChanged, data.name);
                    }
                } catch (err) {
                    logger.error(`'${data.name}' polling error: ${err}`);
                }
                _checkWorking(false);
            } else {
                _checkWorking(false);
            }
        }
    }
    /**
     * Update the Tags values read
     * For WebAPI NotOwn: first convert the request data to a flat struct
     * @param {*} reqdata 
     */
    var _updateVarsValue = (reqdata) => {
        if (!reqdata) return
        const timestamp = new Date().getTime();
        var changed = {};
        if (apiProperty.ownFlag) {
            var newItems = 0;
            for (var i = 0; i < reqdata.length; i++) {
                const id = reqdata[i].id;
                if (id) {
                    if (!data.tags[id]) {
                        newItems++;
                    } else {
                        reqdata[i].daq = data.tags[id].daq;
                    }
                    requestItemsMap[id] = [reqdata[i]];
                    reqdata[i].changed = varsValue[id] && reqdata[i].value !== varsValue[id].value;
                    if (!utils.isNullOrUndefined(reqdata[i].value)) {
                        reqdata[i].value = deviceUtils.tagValueCompose(reqdata[i].value, data.tags[id]);
                        reqdata[i].timestamp = timestamp;
                        if (this.addDaq && deviceUtils.tagDaqToSave(reqdata[i], timestamp)) {
                            changed[id] = reqdata[i];
                        }
                    }
                    reqdata[i].changed = false;
                    varsValue[id] = reqdata[i];
                }
            }
            newItemsCount = newItems;
            return changed;
        } else {
            console.log('apiProperty.ownFlag ==== false')
            var someval = false;
            var result = {};
            var items = _dataToFlat(reqdata, apiProperty);
            for (var key in items) {
                if (requestItemsMap[key]) {
                    for (var index in requestItemsMap[key]) {
                        var tag = requestItemsMap[key][index];
                        if (tag) {
                            someval = true;
                            result[tag.id] = {
                                id: tag.id,
                                value: (tag.memaddress) ? items[tag.memaddress] : items[key],
                                type: items[key].type,
                                daq: tag.daq,
                                tagref: tag
                            };
                        }
                    }
                }
            }
            if (someval) {
                console.log('someval ==== true')
                for (var id in result) {
                    result[id].changed = varsValue[id] && result[id].value !== varsValue[id].value;
                    if (!utils.isNullOrUndefined(result[id].value)) {
                        result[id].value = deviceUtils.tagValueCompose(result[id].value, result[id].tagref);
                        result[id].timestamp = timestamp;
                        if (this.addDaq && deviceUtils.tagDaqToSave(result[id], timestamp)) {
                            changed[id] = result[id];
                        }
                    }
                    result[id].changed = false;
                    varsValue[id] = result[id];
                }
                return changed;
            }
        }
        return null;
    }

    var _checkConnection = function () {
        if (data.property.format) {
            apiProperty.getTags = data.property.address;
            apiProperty.ownFlag = false;
        }
        return true
    }

    this.setValue = function (tagId, value) {
        if (apiProperty.ownFlag && data.tags[tagId]) {
            value = _parseValue(data.tags[tagId].type, value);
            data.tags[tagId].value = deviceUtils.tagRawCalculator(value, data.tags[tagId]);
            axios.post(apiProperty.getTags, [{ id: tagId, value: data.tags[tagId].value }]).then(res => {
                lastTimestampRequest = new Date().getTime();
                logger.info(`setValue '${data.tags[tagId].name}' to ${value})`, true);
            }).catch(err => {
                logger.error(`setValue '${data.tags[tagId].name}' error! ${err}`);
            });

        } else {
            logger.error(`setValue not supported!`, true);
        }
    }

    this.getValue = function (id) {
        if (varsValue[id]) {
            return { id: id, value: varsValue[id].value, ts: lastTimestampValue };
        }
        return null;
    }


    this.disconnect = function () {
        return new Promise((resolve, reject) => {
            //关闭kafka链接
            client.close(function () {
                console.log('Kafka client connection has been closed.');
                _checkWorking(false);
                _emitStatus('connect-off');
                _clearVarsValue();
                resolve(true);
            });
        })
    }

    this.isConnected = function () {
        return true;
    }

    this.getValues = function () {
        return varsValue;
    }

    this.load = function (_data) {
        varsValue = [];
        data = JSON.parse(JSON.stringify(_data));
        try {
            requestItemsMap = {};
            var count = Object.keys(data.tags).length;
            for (var id in data.tags) {
                const address = data.tags[id].address || data.tags[id].id;
                if (!requestItemsMap[address]) {
                    requestItemsMap[address] = [data.tags[id]];
                } else {
                    requestItemsMap[address].push(data.tags[id]);
                }
            }
            logger.info(`'${data.name}' data loaded (${count})`, true);
        } catch (err) {
            logger.error(`'${data.name}' load error! ${err}`);
        }
    }

    this.bindAddDaq = function (fnc) {
        this.addDaq = fnc;
    }

    this.addDaq = null;

    this.getStatus = function () {
        return lastStatus;
    }

    //clear varsValue
    var _clearVarsValue = function () {
        for (var id in varsValue) {
            varsValue[id].value = null;
        }
        _emitValues(varsValue);
    }


    var _emitValues = function (values) {
        events.emit('device-value:changed', { id: data.id, values: values });
    }

    var _resetBrowserTimeout = function () {
        if (timeoutBrowser) {
            clearTimeout(timeoutBrowser);
        }
        timeoutBrowser = setTimeout(() => {
            if (browser && browser.connected) {
                browser.end(true);
            }
        }, 15000);
    }

    var _parseValue = function (type, value) {
        if (type === 'number') {
            return parseFloat(value);
        } else if (type === 'boolean') {
            return Boolean(value);
        } else if (type === 'string') {
            return value;
        } else {
            let val = parseFloat(value);
            if (Number.isNaN(val)) {
                // maybe boolean
                val = Number(value);
                // maybe string
                if (Number.isNaN(val)) {
                    val = value;
                }
            } else {
                val = parseFloat(val.toFixed(5));
            }
            return val;
        }
    }


    /**
 * Emit the kafka client connection status
 * @param {*} status 
 */
    var _emitStatus = function (status) {
        lastStatus = status;
        events.emit('device-status:changed', { id: data.name, status: status });
    }

    this.lastReadTimestamp = function () {
        return lastTimestampValue
    }

    var _checkWorking = function (check) {
        if (check && working) {
            overloading++;
            logger.warn(`'${data.name}' working (connection || polling) overload! ${overloading}`);
            // !The driver don't give the break connection
            if (overloading >= 3) {
                connected = false;
                // disconnect();
            } else {
                return false;
            }
        }
        working = check;
        overloading = 0;
        return true;
    }

    var _dataToFlat = function(data, property) {

        var parseTree = function(nodes, id, parent) {
            let result = {};
            let nodeId = id;
            if (parent) {
                nodeId = parent + ':' + nodeId;
            }
            if (Array.isArray(nodes)) {
                let idx = 0;
                for(var key in nodes) {
                    let tres = parseTree(nodes[key], '[' + idx++ + ']', nodeId);
                    Object.keys(tres).forEach( key => {
                        result[key] = tres[key]; 
                    });
                }
            } else if (nodes && typeof nodes === 'object') {
                for(var key in nodes) {
                    let tres = parseTree(nodes[key], key, nodeId);
                    Object.keys(tres).forEach( key => {
                        result[key] = tres[key]; 
                    });
                }
            } else {
                result[nodeId] = nodes; 
            }
            return result;
        }
        if (property.format === 'JSON') {
            return parseTree(data);
        }
        return data;
    }


    //获取Client 传过来的属性
    function getConnectionOptions(property) {
        return {
            url: property.address,
            format: property.format,
            keepalive: 60,
            connectTimeout: 1 * 1000,
            reconnectPeriod: 0
            //   tls: options.encryption,
            //   clientId: options.clientId,
            //   certValidation: options.certValidation,
            //   subscriptions: options.subscriptions,
            //   certificateAuthority: options.selfSignedCertificate ? options.selfSignedCertificate.data : undefined,
            //   clientCertificate: options.clientCertificate ? options.clientCertificate.data : undefined,
            //   clientKey: options.clientKey ? options.clientKey.data : undefined,
        }
    }
}
module.exports = {
    init: function (settings) {
        // deviceCloseTimeout = settings.deviceCloseTimeout || 15000;
    },
    create: function (data, logger, events) {
        return new KafkaClient(data, logger, events);
    },
}